package com.jml.disruptortest;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


//main函数执行调用
public class DisruptorMain {

    public static void main(String[] args) {
        // 1.创建一个可缓存的线程 提供线程来出发Consumer 的事件处理
        ExecutorService executor = Executors.newCachedThreadPool();
        // 2.创建工厂
        EventFactory<LongEvent> eventFactory = new LongEventFactory();
        // 3.创建ringBuffer 大小
        int ringBufferSize = 1024 * 1024; // ringBufferSize大小一定要是2的N次方
        // 4.创建Disruptor
        //第一个参数是，工厂类对象创建LongEvent，
        // 第二个参数是缓冲区大小要是2的次方，
        // 第三个参数是线程池，利用线程池进行Disrupter内部数据接收处理调度，
        //第四个参数是ProducerType.SINGLE 或者 ProducerType.MULTI,前者表示生产者只有一个，后者表示生产者有多个。
        //第五个参数是一种策略，是做协调的
        //new YieldingWaitStrategy()是性能最好的策略，适用于低延迟的系统，对性能有保证用这个
        //new BlockingWaitStrategy()是最低效的策略，但是对cpu损耗小，在不同的部署坏境中可以提供很强的一致性性能
        //new SleepingWaitStrategy()是最低效的策略，但是对cpu损耗小，对生产者线程影响比较小，适用于异步日志
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory, ringBufferSize, executor,
                ProducerType.SINGLE, new YieldingWaitStrategy());
        // 5.连接消费者--注册消费者(往消费者投入信息)
        //这种模式两个消费者重复消费
        disruptor.handleEventsWith(new LongEventHandler1());
        disruptor.handleEventsWith(new LongEventHandler2());
        // 6.启动
        disruptor.start();
        // 7.创建RingBuffer容器
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        // 8.创建生产者
        LongEventProducer producer = new LongEventProducer(ringBuffer);
        // 9.指定缓冲区大小
        ByteBuffer byteBuffer = ByteBuffer.allocate(8);
        for (int i = 1; i <= 100; i++) {
            byteBuffer.putLong(0, i);
            producer.onData(byteBuffer);
        }
        //10.关闭disruptor和executor
        disruptor.shutdown();
        executor.shutdown();
    }

}

